package docsim;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

public class Map3 extends MapReduceBase
		implements Mapper<LongWritable, Text, Text, Text> {

	private static String dollar = "\\$";
	private static String colon = ":";
	private static String newline = "\n";
	private static String tab = "\t";
	Text docPair;
	Text wordCount;

	@Override
	public void map(LongWritable key, Text value,
			OutputCollector<Text, Text> collector, Reporter arg3)
			throws IOException {
		String[] lines = value.toString().split(newline);
		///System.out.println("lines:"+lines.length);
		for(String line : lines)
		{
			String[] parts = line.toString().split(tab); 
			String word = parts[0];
			//System.out.println("word:"+word);
			String doclist = parts[1];
			String[] split = doclist.split(dollar);
			for (int i = 0; i < split.length; i++) {
				//System.out.println(i+":"+split[i]);
				for (int j = i+1; j < split.length; j++) {
					String[] partsi = split[i].split(colon);
					String[] partsj = split[j].split(colon);
					if(Integer.parseInt(partsi[0]) < Integer.parseInt(partsj[0]))
					{
						docPair = new Text(partsi[0] + colon + partsj[0]);
					}
					else
					{
						docPair = new Text(partsj[0] + colon + partsi[0]);
					}
					wordCount = new Text(word + colon + (Integer.valueOf(partsi[1]) + Integer.valueOf(partsj[1])));
					collector.collect(docPair, wordCount);
					//System.out.println("one map3 output written");
				}
			}
		}
	}
}